#include "../header/fudpReciver.hpp"

fudpReciver::fudpReciver(int _socket_id,struct sockaddr_in& _dst_address, const char* filename) {
    this->socket_id = _socket_id;
    this->dst_address=_dst_address;
    // this->socket_id = _socket_id;
    this->srcName = string(filename);

    this->recBase=0;
    this->recBaseIndex=0;
    this->nextSeqNum=0;
    this->overFlag = 0;
}

bool fudpReciver::recieve() {
    MLOG("begin to recive");

    #pragma omp parallel num_threads(2)
    {
        int tid = omp_get_thread_num();
        int nth = omp_get_num_threads();

        if(tid == 0){
            this->recThread();
        } else {
            this->handleDataThread();
        }
    }

    return true;
}

void fudpReciver::handleDataThread(){
    MLOG("init application handle the data thread");
    msgstream filestream(msgstream::WRITE,this->srcName.c_str());
    int current_window_index = 0;

    while(overFlag != 2){
        // get one
        #pragma omp critical(getBuffer)
        {
            if(this->recBase!=this->nextSeqNum)
                current_window_index = this->recBaseIndex;
            else
                current_window_index = -1;
        }
            
        // handle one
        if(current_window_index != -1){
            datagram_header dheader(this->window[current_window_index]);
            filestream.writePiece(window[current_window_index]+header_size,dheader.dataLength);
        }

        if(current_window_index==-1&&overFlag!=1){
            // avoid read too fast
            usleep(10000);
        }

        // over it
        #pragma omp critical(getBuffer)
        {
            if(current_window_index!= -1) {
                this->recBase++;
                this->recBaseIndex = (this->recBaseIndex+1)%reciver_window_size;
            } else if(overFlag==1){
                overFlag = 2;// close it
            }
        }
    }
}

void fudpReciver::recThread(){
    MLOG("init rec thread");
    struct sockaddr_in addressClient;
    socklen_t clientLen = sizeof(addressClient);
    int rec_size  = 0;
    unsigned long total_size=0;
    char ack_buffer[header_size + 10]={0};
    int current_window_index = 0;

    int no_pkt_count_rtt=0;
    
    // get one
    while(current_window_index==-1||((rec_size = recvfrom(socket_id, 
            window[current_window_index], max_datagram_size,
            0,NULL, NULL))!= -1 || no_pkt_count_rtt <=3)) {
        if(current_window_index!=-1){
            if(rec_size <0) {
                no_pkt_count_rtt++;
                continue;
            } else {
                no_pkt_count_rtt=0;
            }
            
            // MLOG("REC: "<<rec_size);
            datagram_header dheader(this->window[current_window_index]);
            if(dheader.type==type_clientFinish) {
                MLOG("finish rec")
                break;
            } else {
                // MLOG("hadle: "<<size);
                // MLOG("GET:"<<dheader.seqNum<<endl<<"WAITING:"<<this->nextSeqNum);
                
                Datagram dataAck(type_dataAck);

                if(dheader.seqNum==this->nextSeqNum){
                    dataAck.getHeader().seqNum=dheader.seqNum;
                    total_size+=dheader.dataLength;
                }
                else{
                    // MLOG("duplicate pkt");
                    dataAck.getHeader().seqNum=this->nextSeqNum-1;
                    current_window_index = -1;
                    // MLOG("pkt seq:"<<dheader.seqNum);
                }
                dheader.rcv_window_size = reciver_window_size - (this->nextSeqNum - this->recBase);

                int rsize=dataAck.get(ack_buffer);
                if(sendto(this->socket_id,ack_buffer,rsize,0,
                        (struct sockaddr*)&dst_address,sizeof(dst_address))<0) {
                    perror("ERROR in ack send");
                    exit(1);
                } else {
                    // MLOG("ack :"<<dataAck.getHeader().seqNum);
                }
            }
        } else {
            MLOG("NO BUFFER");
            sleep(1);
        }

        #pragma omp critical(getBuffer)
        {
            if(current_window_index != -1){
                this->nextSeqNum ++;
            }
            
            if(this->nextSeqNum - this->recBase < reciver_window_size){
                current_window_index = 
                    (this->nextSeqNum - this->recBase + this->recBaseIndex)%reciver_window_size;
            } else {
                current_window_index = -1;  
            }
        }
    }
    
    this->overFlag = 1;
    if(no_pkt_count_rtt>3){
        MLOG("CLIENT NO RESPONSE");
    }
    
    MLOG("finish rec pkt:"<<this->nextSeqNum-1);
    MLOG("get total size:"<<total_size/1024.0/1024);
}